package ynu.edu.user.linster;

import com.fasterxml.jackson.core.type.TypeReference;
import com.rabbitmq.client.Channel;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;
import ynu.edu.common.constant.RedisConstants;
import ynu.edu.common.domain.stats.StatsData;
import ynu.edu.common.domain.stats.UserGrowthData;
import ynu.edu.common.exception.RedisOperationException;
import ynu.edu.common.handler.RetryHandler;
import ynu.edu.common.utils.RedisCache;
import ynu.edu.user.domain.po.DailyUserStat;
import ynu.edu.user.mapper.UserMapper;

import java.io.IOException;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;


@Slf4j
@Component
@RequiredArgsConstructor
public class UserStatsMessageListener {

    private final RedisCache redisCache;
    private final UserMapper userMapper;

    @Value("${spring.rabbitmq.retry.max-attempts:3}")
    private int maxRetries;

    @RabbitListener(queues = "${rabbitmq.queue-name}",containerFactory = "rabbitListenerContainerFactory")
    public void onMessage(String message, Channel channel,
                          @Header(AmqpHeaders.DELIVERY_TAG) long tag,
                          @Header(name = "x-delivery-count", required = false) Integer deliveryCount) {
        log.info("[UserStatsListener] 收到消息: {}", message);

        boolean success = false;
        try {
            // 1. 查询最近7天注册数据
            List<DailyUserStat> userStats = userMapper.selectUserGrowthIn7Days(); // 每天一条记录
            Long totalCount = userMapper.selectTotalUserCount(); // 总用户数

            // 2. 封装成 UserGrowthData
            UserGrowthData userGrowthData = new UserGrowthData();
            userGrowthData.setDates(
                    userStats.stream().map(DailyUserStat::getDate).collect(Collectors.toList())
            );
            userGrowthData.setCounts(
                    userStats.stream().map(d -> d.getCount().intValue()).collect(Collectors.toList())
            );

            // 3. 包装成 StatsData<UserGrowthData>
            StatsData<UserGrowthData> statsData = StatsData.success(userGrowthData, totalCount);

            // 4. 写入 Redis
            redisCache.setCacheObject(RedisConstants.USER_GROWTH_STATS_KEY, statsData, 24, TimeUnit.HOURS);

            // 5. 校验写入结果
            StatsData<UserGrowthData> cachedObj = redisCache.getCacheObject(
                    RedisConstants.USER_GROWTH_STATS_KEY,
                    new TypeReference<>() {}
            );
            if (cachedObj == null || !cachedObj.equals(statsData)) {
                throw new RedisOperationException("Redis 数据校验失败");
            }

            log.info("[UserStatsListener] 用户统计更新成功，总数: {}", totalCount);
            success = true;

        } catch (RedisOperationException e) {
            log.error("[UserStatsListener] Redis 操作失败", e);
            boolean shouldRequeue = RetryHandler.shouldRetry(deliveryCount, maxRetries, e);
            handleNack(channel, tag, shouldRequeue);
        } catch (Exception e) {
            log.error("[UserStatsListener] 未知异常，进入死信队列", e);
            handleNack(channel, tag, false);
        } finally {
            if (success) {
                handleAck(channel, tag);
            }
        }
    }

    private void handleAck(Channel channel, long tag) {
        try {
            if (channel.isOpen()) {
                channel.basicAck(tag, false);
                log.debug("[UserStatsListener] 消息确认成功, tag: {}", tag);
            } else {
                log.warn("[UserStatsListener] 通道已关闭，无法确认消息, tag: {}", tag);
            }
        } catch (IOException e) {
            log.error("[UserStatsListener] 确认消息失败, tag: {}", tag, e);
        }
    }

    private void handleNack(Channel channel, long tag, boolean requeue) {
        try {
            if (channel.isOpen()) {
                channel.basicNack(tag, false, requeue);
                log.debug("[UserStatsListener] 消息拒绝成功, tag: {}, requeue: {}", tag, requeue);
            } else {
                log.warn("[UserStatsListener] 通道已关闭，无法拒绝消息, tag: {}", tag);
            }
        } catch (IOException e) {
            log.error("[UserStatsListener] 拒绝消息失败, tag: {}", tag, e);
        }
    }
}

